unit Queue;

interface

uses Variants, Windows, SysUtils, Classes, Contnrs, SyncObjs;

type

  TAction = procedure(const data:Variant);

  TWork = class
    action: TAction;
    data: Variant;
    next: TWork;
  end;

  TWorkQueue = class(TObjectQueue)
  protected
    CS: TRTLCriticalSection;
    HasWork: TEvent;
  public
    constructor Create;
    destructor Destroy; override;

    procedure Push(w: TWork);
    function Pop(var w: TWork): Boolean;
  end;

  TWorker = class(TThread)
  public
    procedure Execute; override;
  end;

  TThreadPool = class(TObjectList)
    procedure AddThreads(n:Integer);
    procedure Remove(n:Integer);
    destructor Destroy; override;
  end;

  procedure Send(action:TAction; const data:Variant); overload;
  procedure Send(action:TAction; data:array of Variant); overload;

var
  WorkQueue: TWorkQueue;
  ThreadPool: TThreadPool;

implementation

procedure Send(action:TAction; const data:Variant);
var
  w: TWork;
begin
  w := TWork.Create;
  w.action := action;
  w.data := data;
  WorkQueue.Push(w);
end;

procedure Send(action:TAction; data:array of Variant);
var
  v:Variant;
begin
  v := VarArrayOf(data);
  Send(action, v);
end;

{ TWorkQueue }

constructor TWorkQueue.Create;
begin
  inherited Create;
  InitializeCriticalSection(CS);
  HasWork := TEvent.Create(nil, True, False, 'ActionStream.WorkToDo');
  // HasWork := TEvent.Create('ActionStream.WorkToDo', False);
end;

destructor TWorkQueue.Destroy;
begin
  HasWork.Free;
  DeleteCriticalSection(CS);
end;

procedure TWorkQueue.Push(w: TWork);
begin
  EnterCriticalSection(CS);
    PushItem(w);
    HasWork.SetEvent;
  LeaveCriticalSection(CS);
end;

function TWorkQueue.Pop(var w: TWork): Boolean;
begin
  Result := HasWork.WaitFor(100) = wrSignaled;
  if Result then begin
    EnterCriticalSection(CS);
      Result := Count > 0;
      if Result then
        w := TWork(PopItem)
      else
        HasWork.ResetEvent;
    LeaveCriticalSection(CS);
  end;
end;

{ TWorker }

procedure TWorker.Execute;
var
  w: TWork;
begin
  repeat
    if WorkQueue.Pop(w) then begin
      w.action(w.data);
      w.Free;
    end;
  until Terminated;
end;

{ TThreadPool }

procedure TThreadPool.AddThreads(n: Integer);
var
  i: Integer;
begin
  for i := 1 to n do
    inherited Add(TWorker.Create(False));
end;

procedure TThreadPool.Remove(n: Integer);
var
  i: Integer;
begin
  for i := n-1 downto 0 do begin
    with TThread(Items[i]) do begin
      Terminate;
      WaitFor;
    end;
    Delete(i);
  end;
end;

destructor TThreadPool.Destroy;
begin
  Remove(Count);
  inherited;
end;

initialization
  WorkQueue := TWorkQueue.Create;

  ThreadPool := TThreadPool.Create;
  ThreadPool.AddThreads(3);

finalization
  WorkQueue.Free;
  ThreadPool.Free;
end.
